package com.ww.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object RDDtry03_api_agg {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("try03")
    val sc = new SparkContext(conf)
    val data: RDD[(String, Int)] = sc.parallelize(List(
      ("zhangsan", 234),
      ("zhangsan", 5667),
      ("zhangsan", 343),
      ("lisi", 212),
      ("lisi", 44),
      ("lisi", 33),
      ("wangwu", 535),
      ("wangwu", 22)
    ))
    println("\n======================================groupByKey================================================")

    val group = data.groupByKey()
    group.foreach(println)

    println("\n======================================列转行================================================")
    //(zhangsan,CompactBuffer(234, 5667, 343))
    val group1 = group.flatMap(e => e._2.map(x => (e._1, x)).iterator)
    group1.foreach(println)
    println("\n======================================列转行1================================================")
    //(zhangsan,CompactBuffer(234, 5667, 343))
    val group2 = group.flatMapValues(_.iterator)
    group2.foreach(println)
    println("\n======================================取各个名字的top2================================================")
    //(zhangsan,CompactBuffer(234, 5667, 343))
    group.mapValues(e => e.toList.sorted.take(1)).foreach(println)
    println("\n======================================取各个名字的top2================================================")
    group.flatMapValues(e => e.toList.sorted.take(1)).foreach(println)
    println("\n======================================sum================================================")
    val sum = data.reduceByKey(_ + _)
    sum.foreach(println)
    println("\n======================================max================================================")
    val max = data.reduceByKey((o, n) => if (o > n) o else n)
    max.foreach(println)
    println("\n======================================min================================================")
    val min = data.reduceByKey((o, n) => if (o < n) o else n)
    min.foreach(println)
    println("\n======================================count================================================")
    val count = data.mapValues(e => 1).reduceByKey(_ + _)
    count.foreach(println)
    println("\n======================================avg================================================")
    val avg = sum.join(count).mapValues(e => (e._1 / e._2))
    avg.foreach(println)
    println("\n======================================combine 计算平均值================================================")
    val combine = data.combineByKey(
      //第一个值
      (value: Int) => (value, 1),
      //第二个值
      (o: (Int, Int), n: Int) => (o._1 + n, o._2 + 1),
      //合并
      (v1: (Int, Int), v2: (Int, Int)) => (v1._1 + v2._1, v1._2 + v2._2)
    )
    combine.mapValues(e => e._1 / e._2).foreach(println)
  }
}
